package com.bungarus.processor;

import com.bungarus.busi.DischargeQueue;
import com.bungarus.busi.Discharger;
import com.bungarus.codec.protobuf.MessageProtobuf;
import com.bungarus.model.MessageBuilder;
import com.bungarus.model.MessageProcessException;
import com.bungarus.model.MessageProcessor;
import io.netty.channel.Channel;

/**
 * Created by tang on 2018/12/25.
 */
public class AccessNormalProcessor implements MessageProcessor{
    private static final AccessNormalProcessor processor = new AccessNormalProcessor();
    private AccessNormalProcessor() {
    }

    public static AccessNormalProcessor getInstance() {
        return processor;
    }

    @Override
    public void process(MessageProtobuf.Message message, Channel from, Channel to) throws MessageProcessException {
        if(null != to) {//send message to client and send ack back to busi node
            to.writeAndFlush(message).addListener(channelFuture -> {
                if (channelFuture.isSuccess()) {
                    //send ack message back to busi
                    from.writeAndFlush(MessageBuilder.buildAckMessage(message.getHeader().getId()));
                    //waiting for client's ack
                    DischargeQueue dischargeQueue = Discharger.getQueueInstance(Discharger.DISCHARGE_NAME_CLIENT);
                    dischargeQueue.addToQueue(message);
                }
            });
        }
        else {
            MessageProtobuf.Message offlineMsg = MessageBuilder.buildOfflineMessage(message);
            from.writeAndFlush(offlineMsg).addListener(channelFutrue -> {
                if(channelFutrue.isSuccess()) {
                    from.writeAndFlush(MessageBuilder.buildAckMessage(message.getHeader().getId()));
                    DischargeQueue dischargeQueue = Discharger.getQueueInstance(Discharger.DISCHARGE_NAME_GENERIC);
                    dischargeQueue.addToQueue(offlineMsg);
                }
            });
        }
    }
}
